package lsm

import (
	"errors"
	"fmt"
	"learning/gooop/etcd/raft/logger"
	"learning/gooop/etcd/raft/model"
	"learning/gooop/etcd/raft/roles"
	"learning/gooop/etcd/raft/rpc"
	"learning/gooop/etcd/raft/timeout"
	"sync"
	"time"
)

// tFollowerState presents a follower node
type tFollowerState struct {
	model.TEventDrivenModel

	context    iRaftStateContext
	mInitOnce  sync.Once
	mStartOnce sync.Once

	// update: feLeaderHeartbeat
	mTerm int64

	// update: feInit / feLeaderHeartbeat
	mLeaderHeartbeatTimestamp int64

	// update: feLeaderHeartbeat
	mLeaderID string

	// update: feCandidateRequestVote / feVoteToCandidate
	mLastVotedTerm        int64
	mLastVotedCandidateID string
	mLastVotedTimestamp   int64

	// update: feInit / feDisposing
	mDiseposedFlag bool
}

// trigger: init()
// args: empty
const feInit = "follower.init"

// trigger: Start()
// args: empty
const feStart = "follower.Start"

// trigger: Heartbeat()
// args: *rpc.HeartbeatCmd
const feLeaderHeartbeat = "follower.LeaderHeartbeat"

// trigger: AppendLog()
// args: *rpc.AppendLogCmd
const feLeaderAskingAppendLog = "follower.LeaderAskingAppendLog"

// trigger: whenStartThenBeginWatchLeaderTimeout()
// args: empty
const feLeaderHeartbeatTimeout = "follower.LeaderHeartbeatTimeout"

// trigger: RequestVote()
// args: rpc.RequestVoteCmd
const feCandidateRequestVote = "candidate.RequestVote"

// trigger: RequestVote()
// args: rpc.RequestVoteCmd
const feVoteToCandidate = "follower.CandidateRequestVote"

// trigger: whenLeaderHeartbeatTimeoutThenSwitchToCandidateState
const feDisposing = "follower.Disposing"

func newFollowerState(ctx iRaftStateContext, term int64) IRaftState {
	it := new(tFollowerState)
	it.init(ctx, term)
	return it
}

func (me *tFollowerState) init(ctx iRaftStateContext, term int64) {
	me.mInitOnce.Do(func() {
		me.context = ctx
		me.mTerm = term

		me.logf(".init")
		me.initEventHandlers()
		me.Raise(feInit)
	})
}

func (me *tFollowerState) initEventHandlers() {
	me.logf(".initEventHandlers")

	// write only logic
	me.hookEventsForTerm()
	me.hookEventsForLeaderHeartbeatTimestamp()
	me.hookEventsForLeaderID()
	me.hookEventsForLastVotedTerm()
	me.hookEventsForDisposedFlag()

	// read only logic
	me.Hook(feStart,
		me.whenStartThenBeginWatchLeaderTimeout)
	me.Hook(feLeaderHeartbeatTimeout,
		me.whenLeaderHeartbeatTimeoutThenSwitchToCandidateState)
}

// hookEventsForTerm maintains field: mTerm
// update : feInit / feLeaderHeartbeat
func (me *tFollowerState) hookEventsForTerm() {
	me.Hook(feLeaderHeartbeat, func(e string, args ...interface{}) {
		cmd := args[0].(*rpc.HeartbeatCmd)
		me.mTerm = cmd.Term
		me.logf(".feLeaderHeartbeat, term=%v", cmd.Term)
	})
}

// hookEventsForLeaderHeartbeatClock maintains field: mLeaderHeartbeatClock
// update : feLeaderHeartbeat / feLeaderHeartbeatTimeout
func (me *tFollowerState) hookEventsForLeaderHeartbeatTimestamp() {
	me.Hook(feInit, func(e string, args ...interface{}) {
		me.mLeaderHeartbeatTimestamp = time.Now().UnixNano()
	})

	me.Hook(feLeaderHeartbeat, func(e string, args ...interface{}) {
		me.mLeaderHeartbeatTimestamp = time.Now().UnixNano()
		me.logf(".feLeaderHeartbeat, time=%v", me.mLeaderHeartbeatTimestamp)
	})

	me.Hook(feLeaderHeartbeatTimeout, func(e string, args ...interface{}) {
		me.mLeaderHeartbeatTimestamp = 0
		me.logf(".feLeaderHeartbeatTimeout, time=0")
	})
}

// hookEventsForLeaderID maintains field: mLeaderID
// update : feLeaderHeartbeat / feLeaderHeartbeatTimeout
func (me *tFollowerState) hookEventsForLeaderID() {
	me.Hook(feInit, func(e string, args ...interface{}) {
		me.mLeaderID = gInvalidLeaderID
	})

	me.Hook(feLeaderHeartbeat, func(e string, args ...interface{}) {
		cmd := args[0].(*rpc.HeartbeatCmd)
		if cmd.LeaderID != me.context.Config().GetID() {
			me.mLeaderID = cmd.LeaderID
			me.logf(".feLeaderHeartbeat, leader=%v", cmd.LeaderID)
		}
	})

	me.Hook(feLeaderAskingAppendLog, func(e string, args ...interface{}) {
		cmd := args[0].(*rpc.AppendLogCmd)
		if cmd.LeaderID != me.context.Config().GetID() {
			me.mLeaderID = cmd.LeaderID
			me.logf(".feLeaderAskingAppendLog, leader=%v", cmd.LeaderID)
		}
	})

	me.Hook(feLeaderHeartbeatTimeout, func(e string, args ...interface{}) {
		me.mLeaderID = gInvalidLeaderID
		me.logf(".feLeaderHeartbeatTimeout, leader=null")
	})
}

// hookEventsForLastVotedTerm maintains field: mLastVotedTerm
// update : feCandidateRequestVote / feVoteToCandidate
func (me *tFollowerState) hookEventsForLastVotedTerm() {
	me.Hook(feCandidateRequestVote, func(e string, args ...interface{}) {
		// before voting, check whether last vote timeout
		now := time.Now().UnixNano()
		if time.Duration(now-me.mLastVotedTimestamp)*time.Nanosecond >= fnRandomizeDuration(timeout.ElectionTimeout) {
			// timeout, reset to empty
			me.mLastVotedTerm = 0
			me.mLastVotedCandidateID = ""
			me.mLastVotedTimestamp = 0

			me.logf(".feCandidateRequestVote, reset last vote")
		}
	})

	me.Hook(feVoteToCandidate, func(e string, args ...interface{}) {
		cmd := args[0].(*rpc.RequestVoteCmd)
		me.mLastVotedTerm = cmd.Term
		me.mLastVotedCandidateID = cmd.CandidateID
		me.mLastVotedTimestamp = time.Now().UnixNano()

		me.logf(".feVoteToCandidate, candidate=%v, term=%v", cmd.CandidateID, cmd.Term)
	})
}

// hookEventsForDisposedFlag maintains field: mDisposedFlag
// update: feInit / feDisposing
func (me *tFollowerState) hookEventsForDisposedFlag() {
	me.Hook(feInit, func(e string, args ...interface{}) {
		me.mDiseposedFlag = false
	})

	me.Hook(feDisposing, func(e string, args ...interface{}) {
		me.mDiseposedFlag = true
		me.logf(".feDisposing, disposed=true")
	})
}

func (me *tFollowerState) Start() {
	me.mStartOnce.Do(func() {
		me.logf(".Start")
		me.Raise(feStart)
	})
}

func (me *tFollowerState) whenStartThenBeginWatchLeaderTimeout(e string, args ...interface{}) {
	go func() {
		me.logf(".whenStartThenBeginWatchLeaderTimeout, begin")
		iCheckingTimeoutInterval := fnRandomizeDuration(timeout.HeartbeatTimeout / 3)
		for range time.Tick(iCheckingTimeoutInterval) {
			if me.mDiseposedFlag {
				me.logf(".whenStartThenBeginWatchLeaderTimeout, disposed")
				return
			}

			now := time.Now().UnixNano()
			iHeartbeatTimeoutNanos := fnRandomizeInt64(int64(timeout.HeartbeatTimeout / time.Nanosecond))

			if now-me.mLeaderHeartbeatTimestamp >= iHeartbeatTimeoutNanos {
				me.logf(".whenStartThenBeginWatchLeaderTimeout, raise feLeaderHeartbeatTimeout")
				me.Raise(feLeaderHeartbeatTimeout)
				return
			}
		}
	}()
}

func (me *tFollowerState) whenLeaderHeartbeatTimeoutThenSwitchToCandidateState(_ string, args ...interface{}) {
	me.logf(".whenLeaderHeartbeatTimeoutThenSwitchToCandidateState, term=%v", me.mTerm)

	me.Raise(feDisposing)
	me.context.HandleStateChanged(newCandidateState(me.context, me.mTerm))
}

func (me *tFollowerState) Role() roles.RaftRole {
	return roles.Follower
}

// Heartbeat leader to follower
func (me *tFollowerState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
	// check term
	if cmd.Term < me.mTerm {
		// invalid leader
		ret.Code = rpc.HBTermMismatch
		ret.Term = me.mTerm
		return nil
	}

	// raise LeaderHeartbeat
	me.Raise(feLeaderHeartbeat, cmd)

	// return
	ret.Code = rpc.HBOk
	return nil
}

// AppendLog leader to follower
func (me *tFollowerState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
	ret.Term = me.mTerm

	if cmd.Term < me.mTerm {
		// invalid leader
		ret.Code = rpc.ALTermMismatch
		return nil
	}

	store := me.context.Store()
	entry := cmd.Entry

	// check log: expecting appending action follows previous committing action
	if entry.PrevIndex != store.LastCommittedIndex() || entry.PrevTerm != store.LastCommittedTerm() {
		// check log
		e, log := store.GetLog(entry.Index)
		if e != nil {
			ret.Code = rpc.ALInternalError
			return nil
		}

		if log == nil || log.PrevIndex != entry.PrevIndex || log.PrevTerm != entry.PrevTerm {
			// bad log
			ret.Code = rpc.ALIndexMismatch
			ret.PrevLogIndex = store.LastCommittedIndex()
			ret.PrevLogTerm = store.LastCommittedTerm()
			return nil
		}

		// good log, but old, just ignore it
		ret.Code = rpc.ALOk
		return nil
	}

	// good log
	e := store.Append(entry)
	if e != nil {
		ret.Code = rpc.ALInternalError
		return nil
	} else {
		ret.Code = rpc.ALOk
		return nil
	}
}

// CommitLog leader to follower
func (me *tFollowerState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
	store := me.context.Store()
	if cmd.Index != store.LastAppendedIndex() || cmd.Term != store.LastAppendedTerm() {
		// bad index
		ret.Code = rpc.CLLogNotFound
		return nil
	}

	e := store.Commit(cmd.Index)
	if e != nil {
		ret.Code = rpc.CLInternalError
		return nil
	}

	ret.Code = rpc.CLOk
	return nil
}

// RequestVote candidate to follower
func (me *tFollowerState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
	// before voting
	me.Raise(feCandidateRequestVote, cmd)

	// check term
	if cmd.Term <= me.mTerm {
		ret.Term = me.mTerm
		ret.Code = rpc.RVTermMismatch
		return nil
	}

	// check if already voted another
	if me.mLastVotedTerm >= cmd.Term && me.mLastVotedCandidateID != "" && me.mLastVotedCandidateID != cmd.CandidateID {
		ret.Code = rpc.RVVotedAnother
		return nil
	}

	// check log index
	if cmd.LastLogIndex < me.context.Store().LastCommittedIndex() {
		ret.Code = rpc.RVLogMismatch
		return nil
	}

	// vote ok
	me.Raise(feVoteToCandidate, cmd)
	ret.Term = cmd.Term
	ret.Code = rpc.RVOk

	return nil
}

func (me *tFollowerState) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
	// is leader online?
	leaderID := me.mLeaderID
	if leaderID == gInvalidLeaderID {
		ret.Code = rpc.KVInternalError
		return gErrorLeaderOffline
	}

	// proxy to leader
	return me.context.RaftClientService().UsingKV(leaderID, func(client rpc.IKVStoreRPC) error {
		return client.ExecuteKVCmd(cmd, ret)
	})
}

func (me *tFollowerState) logf(format string, args ...interface{}) {
	prefix := fmt.Sprintf("tFollowerState(%v)", me.context.Config().GetID())
	logger.Logf(prefix+format, args...)
}

var gInvalidLeaderID = ""
var gErrorLeaderOffline = errors.New("leader offline")
